package org.apache.lucene.store;

import static org.apache.lucene.store.GAELuceneConstants.entityKind;
import static org.apache.lucene.store.GAELuceneStaticMethods.checkTransaction;
import static org.apache.lucene.store.GAELuceneStaticMethods.combineMultipleIOExceptions;
import static org.apache.lucene.store.GAELuceneStaticMethods.throwIOWrappedException;

import java.io.IOException;
import java.util.ConcurrentModificationException;
import java.util.NoSuchElementException;
import java.util.logging.Level;
import java.util.logging.Logger;

import com.google.appengine.api.datastore.DatastoreService;
import com.google.appengine.api.datastore.DatastoreServiceFactory;
import com.google.appengine.api.datastore.Entity;
import com.google.appengine.api.datastore.EntityNotFoundException;
import com.google.appengine.api.datastore.Key;
import com.google.appengine.api.datastore.KeyFactory;
import com.google.appengine.api.datastore.Transaction;

public class DatastoreLockFactory extends LockFactory {
	
	public final String directory;
	public final DatastoreService dss = DatastoreServiceFactory.getDatastoreService();
	public final Transaction nullTransaction = null;
	public final Key parentKey;
	
	private static final Logger logger = Logger.getLogger(DatastoreLockFactory.class.getSimpleName());
	
	public DatastoreLockFactory(String directory) {
		this(null, directory);
	}
	
	public DatastoreLockFactory(Key parentKey, String directory) {
		if (directory == null)
			throw new IllegalArgumentException("You must provide a directory name to the DatastoreLockFactory");
		this.directory = directory;
		this.parentKey = parentKey;
	}
	
	private String getPrefixedLockName(String lockName) {
		return GAELuceneStaticMethods.getPrefixedLockName(directory, lockName);
	}
	
	public String getDirectory() {
		return directory;
	}
	
 	private Key getLockKey(String lockName) {
		return KeyFactory.createKey(parentKey, entityKind, lockName);
	}
	
	private void releaseLock(Key key) throws IOException {
		try {
			// delete the key that had been requested
			dss.delete(nullTransaction, key);
		} catch (NoSuchElementException noLockException) {
			// there was no lock with that name to clear, so we are good to go
		} catch (ConcurrentModificationException concurEx) {
			// someone else was trying to do something with it
			// pass this on inside an IOException
			throw new LockReleaseFailedException("There was a concurrency exception while trying to release the lock.");
		} catch (RuntimeException runtimeException) {
			throwIOWrappedException(runtimeException, "deleting lock");
		}
	}
	
	@Override
	public Lock makeLock(String lockName) {
		return new DatastoreLock(getPrefixedLockName(lockName));
	}
	
	public Lock makeLock(String lockName, GAEDirectoryPersister hookDirectory) {
		return new DatastoreLock(getPrefixedLockName(lockName), hookDirectory);
	}
	
	@Override
	public synchronized void clearLock(String lockName) throws IOException {
		releaseLock(getLockKey(getPrefixedLockName(lockName)));
	}

	public class DatastoreLock extends Lock {

		private final String prefixedLockName;
		private final Key key;
		private final GAEDirectoryPersister hookDirectory;
		
		public DatastoreLock(String prefixedLockName) {
			this.prefixedLockName = prefixedLockName;
			this.key = getLockKey(prefixedLockName);
			this.hookDirectory = null;
		}
		
		public DatastoreLock(String prefixedLockName, GAEDirectoryPersister hookDirectory) {
			this.prefixedLockName = prefixedLockName;
			this.key = getLockKey(prefixedLockName);
			this.hookDirectory = hookDirectory;
		}
		
		@Override
		public synchronized boolean isLocked() throws IOException {
			// get setup for retry and exception capturing
			boolean locked = false;
			
			try {			
				// check if there is an entity with this key
				try {
					dss.get(nullTransaction, key);
					locked = true;
				} catch (EntityNotFoundException exception) {}
			} catch (ConcurrentModificationException concurEx) {
				locked = true;
			} catch (RuntimeException runtimeException) {
				throwIOWrappedException(runtimeException, "checking for lock");
			}
		
			return locked;
		}

		@Override
		public synchronized boolean obtain() throws IOException {
			Transaction trx = null;
			Key key2 = null;
			
			// get setup for retry and exception capturing
			boolean success = false;			
			try {
				// start transaction
				trx = dss.beginTransaction();
				// verify there is no entity with this key already
				try {
					// pull the key
					dss.get(trx, key);
					// it worked without an exception, so one already exists
					// therefore this obtain cannot succeed
					success = false;
				} catch (EntityNotFoundException exception) {					
					// get didn't find anything, so lets try to write one
					key2 = dss.put(trx, new Entity(key));
				}
				// check that the key returned and the submitted key matches
				success = key2 != null && key2.equals(key);
				
				// commit or rollback the transaction
				if (success) {
					trx.commit();
					
					// call the hookDirectory pullLatest() method
					try {
						if (hookDirectory != null)
							hookDirectory.pullLatest();
					} catch (IOException pullIOException) {
						// because there was a problem during this, we need to try to clear that lock we just obtained
						// then throw the exception, if we can't do that, we need to make sure the message indicates
						// that this happened.
						try {
							releaseLock(key);
						} catch (IOException releaseIOException) {
							combineMultipleIOExceptions(logger, pullIOException, releaseIOException);
						}
					}
				}
				else
					trx.rollback();
			} catch (ConcurrentModificationException concurEx) {
				// concurrent means another process is trying at the same time
				// neither one can succeed and this exception handler will
				// execute on both of them, first one to try again will hopefully
				// get the lock
				success = false;
			}  catch (RuntimeException runtimeException) {
				throwIOWrappedException(runtimeException, "obtaining lock");
			} finally {
				checkTransaction(trx);
			}
		
			return success;
		}

		@Override
		public synchronized void release() throws IOException {
			IOException pushIOException = null;
			IOException releaseIOException = null;
			
			try {
				if (hookDirectory != null)
					hookDirectory.pushContent();
			} catch (IOException ioException) {
				logger.log(Level.FINE, "pushContent() IOException", ioException);
				pushIOException = new IOException("pushContent() IOException: ", ioException);
			}
			
			try {
				releaseLock(key);
			} catch (IOException ioException) {
				logger.log(Level.FINE, "releaseLock(key) IOException", ioException);
				releaseIOException = new IOException("releaseLock(key) IOException: ", ioException);
			}
			
			combineMultipleIOExceptions(null, pushIOException, releaseIOException);
		}
		
		@Override
		public String toString() {
			return prefixedLockName;
		}
		
	}
	
}
